/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.impl.consumer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.hook.ConsumeMessageContext;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.client.stat.ConsumerStatsManager;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.ThreadFactoryImpl;
import com.alibaba.rocketmq.common.message.MessageConst;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.body.CMResult;
import com.alibaba.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;

/**
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-24
 */
public class ConsumeMessageConcurrentlyService implements ConsumeMessageService {
	private static final Logger log = ClientLogger.getLog();
	private final DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
	private final DefaultMQPushConsumer defaultMQPushConsumer;
	private final MessageListenerConcurrently messageListener;
	private final BlockingQueue<Runnable> consumeRequestQueue;
	private final ThreadPoolExecutor consumeExecutor;
	private final String consumerGroup;

	private final ScheduledExecutorService scheduledExecutorService;

	public ConsumeMessageConcurrentlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl, MessageListenerConcurrently messageListener) {
		this.defaultMQPushConsumerImpl = defaultMQPushConsumerImpl;
		this.messageListener = messageListener;

		this.defaultMQPushConsumer = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer();
		this.consumerGroup = this.defaultMQPushConsumer.getConsumerGroup();
		this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();

		this.consumeExecutor = new ThreadPoolExecutor(//
				this.defaultMQPushConsumer.getConsumeThreadMin(), //
				this.defaultMQPushConsumer.getConsumeThreadMax(), //
				1000 * 60, //
				TimeUnit.MILLISECONDS, //
				this.consumeRequestQueue, //
				new ThreadFactoryImpl("ConsumeMessageThread_"));

		this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));
	}

	public void start() {
	}

	public void shutdown() {
		this.scheduledExecutorService.shutdown();
		this.consumeExecutor.shutdown();
	}

	public ConsumerStatsManager getConsumerStatsManager() {
		return this.defaultMQPushConsumerImpl.getConsumerStatsManager();
	}

	class ConsumeRequest implements Runnable {
		private final List<MessageExt> msgs;
		private final ProcessQueue processQueue;
		private final MessageQueue messageQueue;

		public ConsumeRequest(List<MessageExt> msgs, ProcessQueue processQueue, MessageQueue messageQueue) {
			this.msgs = msgs;
			this.processQueue = processQueue;
			this.messageQueue = messageQueue;
		}

		@Override
		public void run() {
			// 1、检查ProcessQueue.dropped是否为true，若不是则直接返回；
			if (this.processQueue.isDropped()) {
				log.info("the message queue not be able to consume, because it's dropped {}", this.messageQueue);
				return;
			}

			MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
			ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(messageQueue);
			ConsumeConcurrentlyStatus status = null;

			ConsumeMessageContext consumeMessageContext = null;

			// 2、检查DefaultMQPushConsumerImpl.consumeMessageHookList: ArrayList<ConsumeMessageHook>是否为空，
			if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
				// 若不是，则初始化ConsumeMessageContext对象
				consumeMessageContext = new ConsumeMessageContext();
				consumeMessageContext.setConsumerGroup(ConsumeMessageConcurrentlyService.this.defaultMQPushConsumer.getConsumerGroup());
				consumeMessageContext.setMq(messageQueue);
				consumeMessageContext.setMsgList(msgs);
				consumeMessageContext.setSuccess(false);

				/**
				 * 调用ArrayList<ConsumeMessageHook>列表中每个ConsumeMessageHook对象的consumeMessageBefore (ConsumeMessageContext context)方法，
				 * 该consumeMessageHookList变量由业务层调用DefaultMQPushConsumerImpl.registerConsumeMessageHook (ConsumeMessageHook hook)方法设置，
				 * 由业务层自定义ConsumeMessageHook接口的实现类，
				 * 实现该接口的consumeMessageBefore(final ConsumeMessageContext context)和consumeMessageAfter(final ConsumeMessageContext context)方法，
				 * 分别在回调业务层的具体消费方法之前和之后调用者两个方法。
				 */
				ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
			}

			long beginTimestamp = System.currentTimeMillis();

			try {
				// 3、遍历List<MessageExt>列表，检查每个消息（MessageExt对象）中topic值，
				// 若该值等于"%RETRY%+consumerGroup"，则从消息的propertis属性中获取"RETRY_TOPIC"属性的值，
				// 若该属性值不为null，则将该属性值赋值给该消息的topic值；对于重试消息，会将真正的topic值放入该属性中；
				ConsumeMessageConcurrentlyService.this.resetRetryTopic(msgs);

				// 4、调用ConsumeMessageConcurrentlyService.messageListener: MessageListenerConcurrently的consumeMessage方法；
				// 该messageListener变量是由DefaultMQPushConsumer.registerMessageListener (MessageListener messageListener)方法在业务层设置的，
				// 对于并发消费，则在业务层就要实现MessageListenerConcurrently接口的consumeMessage方法；
				// 该回调方法consumeMessage返回ConsumeConcurrentlyStatus. CONSUME_SUCCESS或者ConsumeConcurrentlyStatus.RECONSUME_LATER值；
				status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
			} catch (Throwable e) {
				log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}", //
						RemotingHelper.exceptionSimpleDesc(e), //
						ConsumeMessageConcurrentlyService.this.consumerGroup, //
						msgs, //
						messageQueue);
			}

			long consumeRT = System.currentTimeMillis() - beginTimestamp;

			// 5、上一部分返回值status，若为null，则置status等于ConsumeConcurrentlyStatus.RECONSUME_LATER；
			if (null == status) {
				log.warn("consumeMessage return null, Group: {} Msgs: {} MQ: {}", //
						ConsumeMessageConcurrentlyService.this.consumerGroup, //
						msgs, //
						messageQueue);
				status = ConsumeConcurrentlyStatus.RECONSUME_LATER;
			}

			// 5、检查DefaultMQPushConsumerImpl.consumeMessageHookList: ArrayList<ConsumeMessageHook>是否为空，
			if (ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.hasHook()) {
				// 若不是，则初始化ConsumeMessageContext对象
				consumeMessageContext.setStatus(status.toString());
				consumeMessageContext.setSuccess(ConsumeConcurrentlyStatus.CONSUME_SUCCESS == status);

				// 调用ArrayList<ConsumeMessageHook>列表中每个ConsumeMessageHook对象的executeHookAfter(ConsumeMessageContext context)方法，
				// 该ArrayList<ConsumeMessageHook>列表是应用层设置的回调类；
				ConsumeMessageConcurrentlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
			}

			ConsumeMessageConcurrentlyService.this.getConsumerStatsManager().incConsumeRT(ConsumeMessageConcurrentlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);

			/**
			 * ********************************************************
			 * 6、处理消费失败的消息。
			 * 对于回调方法consumeMessage的执行结果为失败的消息，以内部Producer的名义重发到Broker端用于重试消费。
			 * 若ProcessQueue.dropped为false，调用ConsumeMessageConcurrentlyService. processConsumeResult(ConsumeConcurrentlyStatus status, ConsumeConcurrentlyContext context, ConsumeRequest consumeRequest)方法
			 */
			if (!processQueue.isDropped()) {
				ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
			} else {
				log.warn("processQueue is dropped without process consume result. messageQueue={}, msgTreeMap={}, msgs={}", new Object[]{messageQueue, processQueue.getMsgTreeMap(), msgs});
			}
		}

		public List<MessageExt> getMsgs() {
			return msgs;
		}

		public ProcessQueue getProcessQueue() {
			return processQueue;
		}

		public MessageQueue getMessageQueue() {
			return messageQueue;
		}
	}

	public boolean sendMessageBack(final MessageExt msg, final ConsumeConcurrentlyContext context) {
		int delayLevel = context.getDelayLevelWhenNextConsume();

		try {
			// *********************************************
			this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, context.getMessageQueue().getBrokerName());
			return true;
		} catch (Exception e) {
			log.error("sendMessageBack exception, group: " + this.consumerGroup + " msg: " + msg.toString(), e);
		}

		return false;
	}

	public void processConsumeResult(//
			final ConsumeConcurrentlyStatus status, //
			final ConsumeConcurrentlyContext context, //
			final ConsumeRequest consumeRequest//
	) {
		// 6.1）ConsumeConcurrentlyContext.ackIndex初始的默认值为Integer.MAX_VALUE，
		// 表示消费成功的个数，
		// 该变量值可由业务层在执行回调方法失败之后重新设定；
		int ackIndex = context.getAckIndex();

		if (consumeRequest.getMsgs().isEmpty())
			return;

		switch (status) {
			// 若status等于CONSUME_SUCCESS
			case CONSUME_SUCCESS :
				// 判断ackIndex是否大于ConsumeRequest中MessageExt列表的个数
				if (ackIndex >= consumeRequest.getMsgs().size()) {
					// 如果大于则表示该列表的消息全部消费成功，
					// 则将ackIndex置为List<MessageExt>列表的个数减1，
					ackIndex = consumeRequest.getMsgs().size() - 1;
				}

				// 如果小于则ackIndex
				// 表示消费失败的消息在列表的位置
				// 执行以下操作
				int ok = ackIndex + 1;
				int failed = consumeRequest.getMsgs().size() - ok;
				// 对消费成功个数进行统计
				this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), ok);
				// 对消费失败个数进行统计
				this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), failed);
				break;

			// 若status等于RECONSUME_LATER
			// 则表示消息全部消费失败
			case RECONSUME_LATER :
				// 置ackIndex=-1
				ackIndex = -1;
				// 对消费失败个数进行统计；
				this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), consumeRequest.getMsgs().size());
				break;
			default :
				break;
		}

		switch (this.defaultMQPushConsumer.getMessageModel()) {
			// 6.2）若此消息模式为广播（DefaultMQPushConsumer.messageModel设置），
			case BROADCASTING :
				for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
					// 则从消费失败的列表位置i=（ackIndex+1）开始遍历consumeRequest.getMsgs()即List<MessageExt>列表，
					MessageExt msg = consumeRequest.getMsgs().get(i);
					msg.setConsumeTimestamp(System.currentTimeMillis());
					// 打印消费失败的MessageExt日志信息；
					log.warn("BROADCASTING, the message consume failed, drop it, {}", msg.toString());
				}
				break;

			// 6.3）若此消息模式为集群（DefaultMQPushConsumer.messageModel设置），，
			case CLUSTERING :
				List<MessageExt> msgBackFailed = new ArrayList<MessageExt>(consumeRequest.getMsgs().size());
				for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {

					// 则从消费失败的列表位置（ackIndex+1）开始遍历List<MessageExt>列表
					MessageExt msg = consumeRequest.getMsgs().get(i);

					// ******************************************
					// 对消费失败的信息发送重试消息给Broker（用于消息重试）
					// 对于消费失败的MessageExt对象，调用 ConsumeMessageConcurrentlyService.sendMessageBack (MessageExt msg, ConsumeConcurrentlyContext context)方法向Broker发送CONSUMER_SEND_MSG_BACK请求码；
					boolean result = this.sendMessageBack(msg, context);
					if (!result) {
						// 对于发送BACK消息失败之后发送RETRY消息也失败的MessageExt消息
						msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);

						// 构建一个临时列表msgBackFailedList；
						msgBackFailed.add(msg);
					}
				}

				// 6.4）检查上一步的临时列表msgBackFailedList是否为空，
				// 若不为空则说明有发送重试消息失败的
				if (!msgBackFailed.isEmpty()) {

					// 首先从ConsumeRequest.msgs: List<MessageExt>变量列表中删除消费失败的这部分消息
					consumeRequest.getMsgs().removeAll(msgBackFailed);

					// 将发送BACK消息失败之后发送RETRY消息也失败的List<MessageExt>列表封装成ConsumeRequest线程对象，
					// 再次提交到ConsumeMessageConcurrentlyService.consumeExecutor线程池中；
					this.submitConsumeRequestLater(msgBackFailed, consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue());
				}
				break;
			default :
				break;
		}

		// 6.5）将消费成功了的消息（ConsumeRequest.msgs列表中剩下的）从 ProcessQueue.msgTreeMap列表中删除，
		// 同时更新 ProcessQueue.msgCount并返回该列表中第一个MessageExt元素的key值，即该元素的queueoffset值；
		long offset = consumeRequest.getProcessQueue().removeMessage(consumeRequest.getMsgs());
		if (offset >= 0) {
			// 6.6）若上一步返回的queueoffset大于等于0，则调用 OffsetStore.updateOffset( MessageQueue mq, long offset, boolean increaseOnly)更新该消费队列的消费进度；
			this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), offset, true);
		}
	}

	private void submitConsumeRequestLater(//
			final List<MessageExt> msgs, //
			final ProcessQueue processQueue, //
			final MessageQueue messageQueue//
	) {

		this.scheduledExecutorService.schedule(new Runnable() {

			@Override
			public void run() {
				ConsumeMessageConcurrentlyService.this.submitConsumeRequest(msgs, processQueue, messageQueue, true);
			}
		}, 5000, TimeUnit.MILLISECONDS);
	}

	@Override
	public void submitConsumeRequest(//
			final List<MessageExt> msgs, //
			final ProcessQueue processQueue, //
			final MessageQueue messageQueue, //
			final boolean dispatchToConsume) {
		final int consumeBatchSize = this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();

		// 比较List<MessageExt>列表的个数是否大于了批处理的最大条数
		if (msgs.size() <= consumeBatchSize) {
			// 若没有则以该List<MessageExt>队列、ProcessQueue对象、MessageQueue对象初始化ConsumeMessageConcurrentlyService的内部类 ConsumeRequest 的对象，
			// 并放入ConsumeMessageConcurrentlyService.consumeExecutor线程池中
			ConsumeRequest consumeRequest = new ConsumeRequest(msgs, processQueue, messageQueue);
			this.consumeExecutor.submit(consumeRequest);
		} else {
			// 否则遍历List<MessageExt>列表
			for (int total = 0; total < msgs.size();) {
				// 每次从List<MessageExt>列表中取consumeMessageBatchMaxSize个MessageExt对象构成新的List<MessageExt>列表msgThis
				List<MessageExt> msgThis = new ArrayList<MessageExt>(consumeBatchSize);
				for (int i = 0; i < consumeBatchSize; i++, total++) {
					if (total < msgs.size()) {
						msgThis.add(msgs.get(total));
					} else {
						break;
					}
				}

				// 然后以新的MessageExt队列、ProcessQueue对象、MessageQueue对象初始化ConsumeMessageConcurrentlyService的内部类 ConsumeRequest 的对象
				// 并放入ConsumeMessageConcurrentlyService.consumeExecutor线程池中，直到该队列遍历完为止
				ConsumeRequest consumeRequest = new ConsumeRequest(msgThis, processQueue, messageQueue);
				this.consumeExecutor.submit(consumeRequest);
			}
		}
	}

	@Override
	public void updateCorePoolSize(int corePoolSize) {
		if (corePoolSize > 0 //
				&& corePoolSize <= Short.MAX_VALUE //
				&& corePoolSize < this.defaultMQPushConsumer.getConsumeThreadMax()) {
			this.consumeExecutor.setCorePoolSize(corePoolSize);
		}
	}

	@Override
	public void incCorePoolSize() {
		long corePoolSize = this.consumeExecutor.getCorePoolSize();
		if (corePoolSize < this.defaultMQPushConsumer.getConsumeThreadMax()) {
			this.consumeExecutor.setCorePoolSize(this.consumeExecutor.getCorePoolSize() + 1);
		}

		log.info("incCorePoolSize Concurrently from {} to {}, ConsumerGroup: {}", //
				corePoolSize, //
				this.consumeExecutor.getCorePoolSize(), //
				this.consumerGroup);
	}

	@Override
	public void decCorePoolSize() {
		long corePoolSize = this.consumeExecutor.getCorePoolSize();
		if (corePoolSize > this.defaultMQPushConsumer.getConsumeThreadMin()) {
			this.consumeExecutor.setCorePoolSize(this.consumeExecutor.getCorePoolSize() - 1);
		}

		log.info("decCorePoolSize Concurrently from {} to {}, ConsumerGroup: {}", //
				corePoolSize, //
				this.consumeExecutor.getCorePoolSize(), //
				this.consumerGroup);
	}

	@Override
	public int getCorePoolSize() {
		return this.consumeExecutor.getCorePoolSize();
	}

	@Override
	public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) {
		ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult();
		result.setOrder(false);
		result.setAutoCommit(true);

		List<MessageExt> msgs = new ArrayList<MessageExt>();
		msgs.add(msg);
		MessageQueue mq = new MessageQueue();
		mq.setBrokerName(brokerName);
		mq.setTopic(msg.getTopic());
		mq.setQueueId(msg.getQueueId());

		ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(mq);

		this.resetRetryTopic(msgs);

		final long beginTime = System.currentTimeMillis();

		log.info("consumeMessageDirectly receive new messge: {}", msg);

		try {
			ConsumeConcurrentlyStatus status = this.messageListener.consumeMessage(msgs, context);
			if (status != null) {
				switch (status) {
					case CONSUME_SUCCESS :
						result.setConsumeResult(CMResult.CR_SUCCESS);
						break;
					case RECONSUME_LATER :
						result.setConsumeResult(CMResult.CR_LATER);
						break;
					default :
						break;
				}
			} else {
				result.setConsumeResult(CMResult.CR_RETURN_NULL);
			}
		} catch (Throwable e) {
			result.setConsumeResult(CMResult.CR_THROW_EXCEPTION);
			result.setRemark(RemotingHelper.exceptionSimpleDesc(e));

			log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s", //
					RemotingHelper.exceptionSimpleDesc(e), //
					ConsumeMessageConcurrentlyService.this.consumerGroup, //
					msgs, //
					mq), e);
		}

		result.setSpentTimeMills(System.currentTimeMillis() - beginTime);

		log.info("consumeMessageDirectly Result: {}", result);

		return result;
	}

	public void resetRetryTopic(final List<MessageExt> msgs) {
		// "%RETRY%+consumerGroup"
		final String groupTopic = MixAll.getRetryTopic(consumerGroup);
		// 遍历List<MessageExt>列表
		for (MessageExt msg : msgs) {
			// 从消息的propertis属性中获取"RETRY_TOPIC"属性的值
			String retryTopic = msg.getProperty(MessageConst.PROPERTY_RETRY_TOPIC);
			if (retryTopic != null && groupTopic.equals(msg.getTopic())) {
				// 若该消息的topic为"%RETRY%+consumerGroup"，并且"RETRY_TOPIC"属性的值不为null，
				// 则将该属性值赋值给该消息的topic值；
				// 对于重试消息，会将真正的topic值放入该属性中；
				msg.setTopic(retryTopic);
			}
		}
	}
}
